package com.yh.csx.event.store.gateway.subscribe;

import com.fasterxml.jackson.core.type.TypeReference;
import com.yh.csx.bsf.core.http.DefaultHttpClient;
import com.yh.csx.bsf.core.util.JsonUtils;
import com.yh.csx.event.client.thread.NamedThreadFactory;
import com.yh.csx.event.domain.DataStatusEnum;
import com.yh.csx.event.domain.DelFlagEnum;
import com.yh.csx.event.domain.StoreTypeEnum;
import com.yh.csx.event.domain.gateway.IProducerGateway;
import com.yh.csx.event.store.gateway.project.ProjectRepository;
import com.yh.csx.event.store.gateway.project.dataobject.ProjectConfigDo;
import com.yh.csx.event.store.gateway.subscribe.dataobject.ProducerConfigDo;
import com.yh.csx.event.store.subscribe.dto.resp.ProducerResp;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author zhangsheng
 * @date 2021/8/20 17:33
 */
@Slf4j
@RestController
@RequestMapping("producer")
public class ProducerGateway implements IProducerGateway {
    public static final String CLIENT_MANAGE_SHOW_PRODUCER = "/clientManage/showProducer";
    @Value("${bsf.event.mesh.url:noUrl}")
    private String eventMeshUrl;
    @Value("${bsf.event.pub.timer.delay.seconds:60}")
    private int timerDelaySeconds = 60;
    @Autowired
    private ProjectRepository projectRepository;
    @Autowired
    private ProducerRepository producerRepository;
    @Autowired
    @Qualifier("eventHttpClient")
    private DefaultHttpClient httpClient;
    private ScheduledExecutorService timer;

    @PostConstruct
    public void init() {
        if ("noUrl".equals(eventMeshUrl)) {
            log.info("=loadData=>no url:{}", eventMeshUrl);
            return;
        }
        timer = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("loadProducer", true));
        timer.scheduleWithFixedDelay(() -> loadData(), 10, timerDelaySeconds, TimeUnit.SECONDS);
    }

    @GetMapping("loadData")
    public String loadData() {
        String json = httpClient.get(eventMeshUrl + CLIENT_MANAGE_SHOW_PRODUCER);
        log.debug("=loadData=>json:{}", json);
        List<ProducerResp> list = JsonUtils.deserialize(json, new TypeReference<List<ProducerResp>>() {
        });
        if (CollectionUtils.isEmpty(list)) {
            return null;
        }
        Map<String, ProjectConfigDo> configMap = projectRepository.findAll().stream()
                .collect(Collectors.toMap(ProjectConfigDo::getProjectBeName, Function.identity()));

        List<ProducerConfigDo> configDoList = list.stream()
                .map(resp -> resp.getTopics().stream().map(topic -> createProducerConfig(resp, configMap, topic))
                                .filter(Objects::nonNull)
                                .collect(Collectors.toList()))
                .filter(Objects::nonNull)
                .flatMap(Collection::stream)
                .collect(Collectors.toList());

        Map<String, ProducerConfigDo> storedMap = producerRepository.findAll().stream()
                .collect(Collectors.toMap(ProducerConfigDo::getTopic, Function.identity()));

        producerRepository.saveAll(configDoList.stream()
                .filter(pc -> null == storedMap.get(pc.getTopic()))
                .collect(Collectors.toList()));

        Map<String, ProducerConfigDo> update = configDoList.stream()
                .filter(pc -> null != storedMap.get(pc.getTopic()))
                .collect(Collectors.toMap(ProducerConfigDo::getTopic, Function.identity()));

        storedMap.values().stream()
                .filter(f -> null != update.get(f.getTopic()))
                .forEach(f -> {
                    f.setAppName(update.get(f.getTopic()).getAppName());
                    f.setProjectId(update.get(f.getTopic()).getProjectId());
                    f.setRemark(update.get(f.getTopic()).getRemark());
                    f.setUpdateTime(new Date());
                });

        producerRepository.flush();
        return JsonUtils.serialize(list);
    }

    private ProducerConfigDo createProducerConfig(ProducerResp resp, Map<String, ProjectConfigDo> map, String topic) {
        if (null == map.get(resp.getAppName())) {
            log.error("=createProducerConfig=>appName:{}", resp.getAppName());
            return null;
        }
        ProducerConfigDo ds = new ProducerConfigDo();
        ds.setProjectId(map.get(resp.getAppName()).getId());
        ds.setTopic(topic);
        ds.setAppName(resp.getAppName());
        ds.setTopicName("sync");
        ds.setCreator("sync");
        ds.setUpdater(ds.getCreator());
        ds.setDelFlag(DelFlagEnum.NORMAL.getCode());
        ds.setStatus(DataStatusEnum.NORMAL.getCode());
        ds.setStoreType(StoreTypeEnum.ROCKET_MQ.getCode());
        ds.setRemark(resp.getHost());
        ds.setCreateTime(new Date());
        ds.setUpdateTime(new Date());
        return ds;
    }

    @PreDestroy
    public void destroy() {
        if (null != timer) {
            timer.shutdown();
        }
    }
}
